package realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import realtime.app.dwd.db.DwdUserRegister;
import realtime.app.func.BeanToJsonStrFunction;
import realtime.beans.UserRegisterBean;
import realtime.util.DateFormatUtil;
import realtime.util.DorisUtil;
import realtime.util.EnvUtil;
import realtime.util.KafkaUtil;

import java.util.Date;

public class DwsUserUserAddWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = EnvUtil.getSEE(4);
        String topic = "dwd_user_register";
        String groupId = "dws_user_user_add_window";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaStrDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        //对流中数据转化为JSONobj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(jsonStr);
                        out.collect(jsonObject);
                    }
                }
        );
        //指定水位线
        SingleOutputStreamOperator<JSONObject> watermarkDS = jsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject jsonObject, long l) {
                                return jsonObject.getLong("ts") * 1000;
                            }
                        })
        );
        //按照用户分组
        KeyedStream<JSONObject, String> keyedDS =
                watermarkDS.keyBy(v1 -> v1.getString("id"));
        //对数据进行处理并转化为实体类
        SingleOutputStreamOperator<UserRegisterBean> processDS = keyedDS.process(new KeyedProcessFunction<String, JSONObject, UserRegisterBean>() {
            @Override
            public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, UserRegisterBean>.Context ctx, Collector<UserRegisterBean> out) throws Exception {
                UserRegisterBean userRegisterBean = new UserRegisterBean(
                        "",
                        "",
                        "",
                        1L
                );
                out.collect(userRegisterBean);
            }
        });
        //开窗
        AllWindowedStream<UserRegisterBean, TimeWindow> windowDS = processDS.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));
        //聚合
        SingleOutputStreamOperator<UserRegisterBean> reduceDS = windowDS.reduce(
                new ReduceFunction<UserRegisterBean>() {
                    @Override
                    public UserRegisterBean reduce(UserRegisterBean v1, UserRegisterBean v2) throws Exception {
                        v1.setRegisterCt(v1.getRegisterCt() + v2.getRegisterCt());
                        return v1;
                    }
                },
                new AllWindowFunction<UserRegisterBean, UserRegisterBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<UserRegisterBean> values, Collector<UserRegisterBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        long timeMillis = System.currentTimeMillis();
                        String curDate = DateFormatUtil.toDate(timeMillis);
                        for (UserRegisterBean value : values) {
                            value.setStt(stt);
                            value.setEdt(edt);
                            value.setCurDate(curDate);
                            out.collect(value);
                        }
                    }
                }
        );
        //reduceDS.print(">>>>");
        reduceDS
                .map(new BeanToJsonStrFunction<>())
                .sinkTo(DorisUtil.getDorisSink("dws_user_user_add_window"));
        env.execute();
    }
}
